package me.shiqiang.im;

import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.corundumstudio.socketio.Configuration;
import com.corundumstudio.socketio.SocketConfig;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.protocol.JacksonJsonSupport;
import me.shiqiang.im.config.RocketMqConsumerConfig;
import me.shiqiang.im.config.RocketMqProducerConfig;
import me.shiqiang.im.linsteners.Listener;
import me.shiqiang.im.server.Auth;
import me.shiqiang.im.util.RedisPipelineProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.cloud.client.SpringCloudApplication;
import org.springframework.cloud.netflix.feign.EnableFeignClients;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.EnableScheduling;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;

@SpringCloudApplication
@EnableFeignClients
@EnableCaching
@EnableAutoConfiguration
@EnableScheduling
public class ImSocketioServerApplication {

	public static void main(String[] args) {
		SpringApplication.run(ImSocketioServerApplication.class, args);
	}

    @Bean
    public ConcurrentLinkedQueue<RedisPipelineProcessor> createQueue() {
        return new ConcurrentLinkedQueue<>();
    }


    @Value("${connect.port}")
    String port;

    @Autowired
    private Map<String,Listener> listenerMap;

	@Autowired
	private Map<String,MessageListener> consumerMap;

    @Resource
    private Auth auth;

    @Value("#{${com.tst.rocketmq.consumer.subscription}}")
    Map<String, String> consumerSubscription;


    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public ConsumerBean createConsumerBean(RocketMqConsumerConfig config) {
        ConsumerBean consumerBean = new ConsumerBean();
        consumerBean.setProperties(config.getProperties());
        Map<Subscription, MessageListener> table = new HashMap<>();
        for (Map.Entry<String, MessageListener> entry : consumerMap.entrySet()) {
            String[] keys = consumerSubscription.get(entry.getKey()).split("#");
            Subscription subscription = new Subscription();
            subscription.setTopic(keys[0]);
            subscription.setExpression(keys[1]);
            table.put(subscription, entry.getValue());
        }
        consumerBean.setSubscriptionTable(table);
        return consumerBean;
    }

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public ProducerBean createProducerBean(RocketMqProducerConfig config) {
        ProducerBean producerBean = new ProducerBean();
        producerBean.setProperties(config.getProperties());
        return producerBean;
    }



	@Bean(initMethod = "start",destroyMethod = "stop")
	public SocketIOServer createServer(){
		Configuration configuration = new Configuration();
//        configuration.setAuthorizationListener(auth);
//		configuration.setHostname("192.168.20.143");
		configuration.setPort(Integer.parseInt(port));
        SocketConfig socketConfig = configuration.getSocketConfig();
        socketConfig.setReuseAddress(true);//添加此配置，解决重启报 Address already in use异常
        JacksonJsonSupport jacksonJsonSupport = new JacksonJsonSupport();
        configuration.setJsonSupport(jacksonJsonSupport);

		SocketIOServer server = new SocketIOServer(configuration);
        for (Listener listener : listenerMap.values()) {
            listener.setServer(server);
            server.addListeners(listener);
        }
		return server;
	}
}
